package com.aotain.project.apollo.topo;

import org.apache.log4j.Logger;
import org.apache.storm.kafka.KafkaSpout;
import org.apache.storm.kafka.KafkaSpout2;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

import com.aotain.common.ZkKafkaConfig;
import com.aotain.kafka.KafkaConfigUtil;
import com.aotain.kafka.KafkaProperties;
import com.aotain.apollo.ApolloConfig;
import com.aotain.apollo.ApolloConfig_Oracle;
import com.aotain.project.apollo.bolt.AbnReverseStatisBolt;
import com.aotain.project.apollo.bolt.AbnSourceStatisBolt;
import com.aotain.project.apollo.bolt.AbnStatisBolt;
import com.aotain.project.apollo.bolt.AbnormalIntoHbaseBolt;
import com.aotain.project.apollo.bolt.BaseStatSumBoltDelay;
import com.aotain.project.apollo.bolt.DosHurstBolt;
import com.aotain.project.apollo.bolt.HurstBolt;
import com.aotain.project.apollo.bolt.HurstFmtBoltTest;
import com.aotain.project.apollo.bolt.IDCStatisBolt;
import com.aotain.project.apollo.bolt.PortAbnStatisBolt;
import com.aotain.project.apollo.bolt.PortDetectBolt;
import com.aotain.project.apollo.bolt.SQLInjectionBolt;
import com.aotain.project.apollo.bolt.SdsSessStat5MinBolt;
import com.aotain.project.apollo.bolt.SessSourBolt;
import com.aotain.project.apollo.bolt.UDIIAbnormalCalBolt;
import com.aotain.project.apollo.bolt.UDIIAbnormalCalReverseBolt;
import com.aotain.project.apollo.bolt.UDIIBaseStatSumBolt;
import com.aotain.project.apollo.bolt.UDIIBaseStatSumSourceBolt;
import com.aotain.project.apollo.bolt.UDIISourceFmtBolt;
import com.aotain.project.apollo.boltv2.ACKFloodBolt;
import com.aotain.project.apollo.boltv2.CCDetectBoltV2;
import com.aotain.project.apollo.boltv2.DNSAttackBolt;
import com.aotain.project.apollo.boltv2.SYNFloodBolt;
import com.aotain.project.apollo.boltv2.UDPFloodBolt;
import com.aotain.project.apollo.boltv2.XSSDetectBolt;

public class DosHurstTopo {
	
	static int workersNum = 6;
	static int spoutNum = 12;
	static int boltNum = 6;
	static int sourcefmtBoltNum = 6;
	public static void main(String[] args) {  
   
		if (args.length != 5){
			   System.err.printf("Usage:<TopoName> <Workers> <SpoutThreadNum> <BoltThreadNum> <SourceFmtboltThreadNum>");
			   System.exit(1);
			}  
		
		workersNum = Integer.parseInt(args[1]);
		spoutNum = Integer.parseInt(args[2]);
		boltNum = Integer.parseInt(args[3]);
		sourcefmtBoltNum = Integer.parseInt(args[4]);
		
		ZkKafkaConfig z_kConfig = new ZkKafkaConfig("/home/storm/config/dbconfig.ini");
		Logger.getLogger(UDIIAbnormalLogCheckTopo.class).info("Version 20160303");
		
		ApolloConfig ap = new ApolloConfig();
		ApolloConfig_Oracle ap_ = new ApolloConfig_Oracle();
		
		String dbJson = ap.getDBObjectJson();
		String dbJson_ = ap_.getDBObjectJson();
		Logger.getLogger(UDIIAbnormalLogCheckTopo.class).info(dbJson);
		
		TopologyBuilder builder = new TopologyBuilder();
		builder.setSpout("UDIIsourceSpout", new KafkaSpout2(new KafkaConfigUtil(z_kConfig.getZkConnServer(),KafkaProperties.UDII_Abnormal_Log_topic,KafkaProperties.zkRoot_UDII,KafkaProperties.Log_UDII_Abnormal_groupId ).getConfig()),spoutNum);
		builder.setBolt("UDIISourceFmtBolt", new UDIISourceFmtBolt(),sourcefmtBoltNum).shuffleGrouping("UDIIsourceSpout");
		builder.setBolt("SQLInjectionBolt", new SQLInjectionBolt(),boltNum).fieldsGrouping("UDIISourceFmtBolt","DipStreamId", new Fields("dip"));
		
		Config conf = new Config();    
		conf.setNumWorkers(workersNum); 
		conf.put("mushroomserver", z_kConfig.getMushroomServer());
		conf.put("dbjson", dbJson);
		conf.put("dbjson_", dbJson_);
		conf.put("sqlinjection", z_kConfig.getSQLInjection());
		Logger.getLogger(UDIIAbnormalLogCheckTopo.class).info(dbJson);
		conf.put("zkConn", z_kConfig.getZkConnServer());
		conf.setNumAckers(0);
		conf.setDebug(false);
		
		if (args.length > 0) {  
			try {
					StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
			} catch (AlreadyAliveException e) {
				e.printStackTrace();
			} catch (InvalidTopologyException e) {
				e.printStackTrace();
			}	catch (AuthorizationException e) {
				e.printStackTrace();
			}
		}else {
			LocalCluster localCluster = new LocalCluster();
			localCluster.submitTopology("AbnormalLogTopo", conf, builder.createTopology());
		}
	}
	
	
//	public static void main(String[] args) {
//		//String toponame = "DDosHurstTopo";
//		
//		ZkKafkaConfig z_kConfig = new ZkKafkaConfig("/home/storm/config/dbconfig.ini");
//		
//		
//		TopologyBuilder builder = new TopologyBuilder();
//		builder.setSpout("UDIIsourceSpout", new KafkaSpout(new KafkaConfigUtil(z_kConfig.getZkConnServer(),"udii","/TestLog",KafkaProperties.Log_Abnormal_groupId2 ).getConfig()),6);
//		builder.setBolt("UDIISourceFmtBolt", new HurstFmtBoltTest(),6).shuffleGrouping("UDIIsourceSpout");
//		
//		if(args[0].equals("CalHurstTest")){
//			builder.setBolt("CalHurstTest", new HurstBolt(),6).fieldsGrouping("UDIISourceFmtBolt","UDIIStreamId", new Fields("dip"));
//		}else{
//			builder.setBolt("DosHurstBolt", new DosHurstBolt(),6).fieldsGrouping("UDIISourceFmtBolt","UDIIStreamId", new Fields("dip"));
//		}
//	
//		
//		Config conf = new Config();
//		conf.setNumWorkers(3);
//		conf.setNumAckers(0);
//		conf.setDebug(false);
//		conf.put("mushroomserver", z_kConfig.getMushroomServer());
//		conf.put("zkConn", z_kConfig.getZkConnServer());
//		
//		if (args.length > 0) {  
//			try {
//				StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
//			} catch (AlreadyAliveException e) {
//				e.printStackTrace();
//			} catch (InvalidTopologyException e) {
//				e.printStackTrace();
//			}	catch (AuthorizationException e) {
//				// TODO Auto-generated catch block
//				e.printStackTrace();
//			}
//		}else {
//			LocalCluster localCluster = new LocalCluster();
//			localCluster.submitTopology("DDosHurstTopo", conf, builder.createTopology());
//		}
//	}
}
